package com.atguigu.tingshu.search.listener;

import com.atguigu.tingshu.common.constant.RabbitConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.execption.GuiguException;
import com.atguigu.tingshu.common.result.ResultCodeEnum;
import com.atguigu.tingshu.model.search.AlbumInfoIndex;
import com.atguigu.tingshu.search.repository.AlbumInfoIndexRepository;
import com.atguigu.tingshu.search.service.SearchService;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.client.elc.ElasticsearchTemplate;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.io.IOException;
import java.math.BigDecimal;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.TimeUnit;

@Component
@Slf4j
public class SearchMqListener {

    @Autowired
    SearchService searchService;
    @Autowired
    RedisTemplate redisTemplate;
    //    @Autowired
//    ElasticsearchTemplate elasticsearchTemplate;
    @Autowired
    AlbumInfoIndexRepository albumInfoIndexRepository;

    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(value = RabbitConstant.ALBUM_UPDATE_SALES_QUEUE),
                    exchange = @Exchange(value = RabbitConstant.ALBUM_EXCHANGE,
                            type = ExchangeTypes.TOPIC),
                    key = {RabbitConstant.ALBUM_UPDATE_SALES_RK}
            )
    )
    public void updateAlbumSales(Channel channel, Message message,
                                 Map<Long, Integer> map) throws IOException {
        try {
            if (!CollectionUtils.isEmpty(map)) {
                //1、根据专辑id查询专辑

                map.forEach((albumId, saleNum) -> {
//                     elasticsearchTemplate.search( , AlbumInfoIndex.class ,)
                    Optional<AlbumInfoIndex> optional = albumInfoIndexRepository.findById(albumId);
                    if (optional.isPresent()) {
                        AlbumInfoIndex albumInfoIndex = optional.get();
                        albumInfoIndex.setBuyStatNum(albumInfoIndex.getBuyStatNum() + saleNum);
                        BigDecimal hotScore =
                                new BigDecimal("0.4")
                                        .multiply(new BigDecimal(albumInfoIndex.getBuyStatNum()))
                                        .add(
                                                new BigDecimal("0.3")
                                                        .multiply(new BigDecimal(albumInfoIndex.getSubscribeStatNum()))
                                        )
                                        .add(
                                                new BigDecimal("0.2")
                                                        .multiply(new BigDecimal(albumInfoIndex.getCommentStatNum()))
                                        )
                                        .add(
                                                new BigDecimal("0.1")
                                                        .multiply(new BigDecimal(albumInfoIndex.getPlayStatNum()))
                                        );
                        albumInfoIndex.setHotScore(hotScore.doubleValue());
                        //更新到es中
                        albumInfoIndexRepository.save(albumInfoIndex);//有id表示更新
                    }
                });
            }
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            if (message.getMessageProperties().isRedelivered()) {
                //丢弃消息
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
                throw new GuiguException(ResultCodeEnum.DATA_ERROR);
            } else {
                //第一次消费异常时需要重试，可以删除redis幂等性token
                channel.basicNack(message.getMessageProperties().getDeliveryTag(),
                        false, true);
            }
        }
    }

    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(value = RabbitConstant.ALBUM_UPPER_QUEUE),
                    exchange = @Exchange(value = RabbitConstant.ALBUM_EXCHANGE, type = ExchangeTypes.TOPIC),
                    key = {RabbitConstant.ALBUM_UPPER_RK}
            )
    )
    public void albumUpper(Channel channel, Message message, Long albumId) throws IOException {
        try {
            //幂等性： albumId
            // redis+token
            Boolean flag = redisTemplate.opsForValue().setIfAbsent(RedisConstant.ALBUM_INFO_PREFIX + albumId
                    , 1, 10, TimeUnit.MINUTES);
            if (!flag) {
                return;
            }
            searchService.upperAlbum(albumId);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            if (message.getMessageProperties().isRedelivered()) {
                log.error("专辑上架异常，专辑id:{}", albumId);
                //丢弃消息
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
                throw new GuiguException(ResultCodeEnum.DATA_ERROR);
            } else {
                //第一次消费异常时需要重试，可以删除redis幂等性token
                redisTemplate.delete(RedisConstant.ALBUM_INFO_PREFIX + albumId);
                channel.basicNack(message.getMessageProperties().getDeliveryTag(),
                        false, true);
            }
        }
    }


    //专辑下架
    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(value = RabbitConstant.ALBUM_DOWN_QUEUE),
                    exchange = @Exchange(value = RabbitConstant.ALBUM_EXCHANGE, type = ExchangeTypes.TOPIC),
                    key = {RabbitConstant.ALBUM_DOWN_RK}
            )
    )
    public void albumDown(Channel channel, Message message, Long albumId) throws IOException {
        try {
            searchService.deleteAlbumInfoIndex(albumId);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            if (message.getMessageProperties().isRedelivered()) {
                log.error("专辑上架异常，专辑id:{}", albumId);
                //丢弃消息
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
                throw new GuiguException(ResultCodeEnum.DATA_ERROR);
            } else {
                //第一次消费异常时需要重试，可以删除redis幂等性token
                redisTemplate.delete(RedisConstant.ALBUM_INFO_PREFIX + albumId);
                channel.basicNack(message.getMessageProperties().getDeliveryTag(),
                        false, true);
            }
        }
    }
}
